package com.zdp.rocketmq.send;

import com.zdp.rocketmq.config.JmsConfig;
import com.zdp.rocketmq.producer.ProducerBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.HashMap;
import java.util.List;

/**
 * @author sesshomaru
 * @date 2021/6/21
 * 指定队列投递
 */
@Slf4j
@RestController
public class ProducerMessageQueueSelectorController {

    @Autowired
    private ProducerBuilder producerBuilder;

    @RequestMapping("/api/v1/send/queue/selector")
    public Object send(String keys, String tag) throws InterruptedException, RemotingException, MQClientException, MQBrokerException {
        Message message = new Message(JmsConfig.GENERAL_TOPIC, tag, keys, ("hello rocketmq = " + keys).getBytes());
        SendResult sendResult = producerBuilder.getProducer().send(message, new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                int queueNum = Integer.parseInt(String.valueOf(arg));
                return mqs.get(queueNum);
            }
            // 这里指定1就是投递到queue-1上，如果超过topic下最大的queue长度，就会报错java.lang.IndexOutOfBoundsException: Index: 5, Size: 4
        }, 2);
        log.info("发送普通消息:{}", sendResult.toString());
        return new HashMap<>(1);
    }
}
